package com.lagou.client;

import com.lagou.service.RpcRequest;
import org.apache.curator.framework.CuratorFramework;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;

public class RpcConsumer {

    //创建线程池对象
    private static final ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

    private Map<String, Invoke> invokeMap = new HashMap<>();

    private CuratorFramework client;

    private String pPath;

    public void setClient(CuratorFramework client) {
        this.client = client;
    }

    public void setpPath(String pPath) {
        this.pPath = pPath;
    }

    //1.创建一个代理对象 providerName：UserService#sayHello are you ok?
    public Object createProxy(final Class<?> serviceClass) {
        //借助JDK动态代理生成代理对象
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(), new Class<?>[]{serviceClass}, new InvocationHandler() {
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                //（1）调用初始化netty客户端的方法
                //封装
                RpcRequest request = new RpcRequest();
                String requestId = UUID.randomUUID().toString();
                System.out.println(requestId);

                String className = method.getDeclaringClass().getName();
                String methodName = method.getName();

                Class<?>[] parameterTypes = method.getParameterTypes();

                request.setRequestId(requestId);
                request.setClassName(className);
                request.setMethodName(methodName);
                request.setParameterTypes(parameterTypes);
                request.setParameters(args);

                if (!invokeMap.isEmpty()){
                    TreeMap<Long, List<Invoke>> elapsedMap = invokeMap.values().stream().collect(Collectors.groupingBy(Invoke::getElapsed, TreeMap::new,Collectors.toList()));
                    Map.Entry<Long, List<Invoke>> longListEntry = elapsedMap.firstEntry();
                    Optional<Invoke> optional = longListEntry.getValue().stream().findAny();
                    Invoke invoke = optional.get();
                    System.out.println("当前host:" + invoke.getHost());
                    // 设置参数
                    invoke.getUserClientHandler().setPara(request);
                    System.out.println(request);
                    System.out.println("设置参数完成");

                    // 去服务端请求数据
                    long start = System.currentTimeMillis();
                    Object result = executor.submit(invoke.getUserClientHandler()).get();
                    long end = System.currentTimeMillis();
                    long elapsed = end - start;
                    if (elapsed > 5000){
                        client.setData().forPath(pPath + "/" + invoke.getHost(),"-1".getBytes());
                    } else {
                        client.setData().forPath(pPath + "/" + invoke.getHost(),(start + ":" + elapsed).getBytes());
                    }
                    return result;
                }
                return null;
            }
        });
    }

    //2.初始化netty客户端
    public synchronized void initClient(List<String> hosts) throws Exception {
        for (String host : hosts) {
            Invoke invoke = new Invoke();
            invoke.initClient(host);
            Long elapsed = getElapsed(host);
            if (elapsed != -1){
                invoke.setElapsed(elapsed);
                invoke.setHost(host);
                invokeMap.put(host, invoke);
            }
        }
    }

    //2.初始化netty客户端
    public synchronized void initClient(String host) throws Exception {
        Long elapsed = getElapsed(host);
        if (!invokeMap.containsKey(host)){
            Invoke invoke = new Invoke();
            invoke.initClient(host);
            if (elapsed != -1){
                invoke.setElapsed(elapsed);
                invoke.setHost(host);
                invokeMap.put(host, invoke);
            }
        } else {
            Invoke invoke = invokeMap.get(host);
            if (elapsed != -1){
                invoke.setElapsed(elapsed);
            } else {
                this.invokeMap.remove(host);
            }
        }
    }

    private Long getElapsed(String host) throws Exception {
        byte[] bytes = client.getData().forPath(pPath + "/" + host);
        String data = new String(bytes);
        if (data.equals("-1")){
            return  -1L;
        } else if (data.equals("0")){
            return  0L;
        } else {
            return Long.parseLong(data.split(":")[1]);
        }
    }

    public synchronized void close(String host) throws InterruptedException {
        Invoke invoke = invokeMap.get(host);
        invoke.close();
        invokeMap.remove(host);
    }

    public synchronized void reset() throws Exception {
        Long now = System.currentTimeMillis() - 5000;
        for (Invoke value : invokeMap.values()) {
            byte[] bytes = client.getData().forPath(pPath + "/" + value.getHost());
            String data = new String(bytes);
            if (!data.equals("-1") && !data.equals("0")){
                Long reqTime = Long.parseLong(data.split(":")[0]);
                if (now > reqTime){
                    client.setData().forPath(pPath + "/" + value.getHost(),"0".getBytes());
                }
            }
        }
    }

}